package com.danbay.framework.kafka;

import com.danbay.framework.utils.RandomUtils;
import com.danbay.framework.utils.StringUtils;
import org.joda.time.DateTime;

import java.util.concurrent.atomic.AtomicLong;

/**
 * 消息序号
 *
 * @author SJG
 *         2017/11/13.
 */
public class MessageSequence {

    private static String lastIpPart = "1";
    private static int nowSecond=1;
    private static int maxSequence=100000;

    static {
        //取得本机ip最后一段
        lastIpPart = RandomUtils.getRandNum(99)+"";
        //取当前时间秒
        nowSecond= DateTime.now().getSecondOfDay();
    }

    private static AtomicLong sequenceAto = new AtomicLong(0);

    /**
     * 取下一个消息序号
     * @return 序号
     */
    public static Long getNextSequence() {
        //以ip最后一段+启动程序时的秒 为前缀
        String prefix=lastIpPart+nowSecond;
        int idLen=15;
        String id = StringUtils.rightPad(prefix, idLen, '0');
        Long increaseId=sequenceAto.incrementAndGet();
        Long sequence = Long.parseLong(id) + increaseId;
        //例id:101532690000003
        //如果sequence大于10万从1开始
        if(increaseId>=maxSequence){
            sequenceAto.compareAndSet(increaseId,0);
        }
        return sequence;
    }
}
